package org.ehe.business.overdue.queue;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.ehe.business.timeout.domain.TimeoutInstance;
import org.ehe.business.timeout.event.RemindEvent;
import org.ehe.business.timeout.event.TimeoutEvent;
import org.ehe.business.timeout.service.TimeoutInstanceService;
import org.ehe.common.core.utils.StringUtils;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.ZoneOffset;
import java.util.*;
import java.util.stream.Collectors;

/**
 * Redis延迟队列实现
 * @author :zhangnn
 * @className :RedisDelayQueue
 * @description: TODO
 * @date 2025-07-29 15:23:13
 */

@Component
@Slf4j
public class RedisDelayQueue {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    private TimeoutInstanceService timeoutInstanceService;
    @Autowired
    public void setTimeoutInstanceService(TimeoutInstanceService timeoutInstanceService) {
        this.timeoutInstanceService = timeoutInstanceService;
    }


    private static final String TIMEOUT_QUEUE_KEY = "timeout:tasks";
    private static final String REMIND_QUEUE_KEY = "timeout:remind";
    // 任务映射键名（用于快速删除）
    private static final String TASK_MAPPING_KEY = "timeout:mapping:";


    private static final String OVERDUE_TASK_KEY = "overdue:tasks";
    private static final String OVERDUE_REMIND_KEY = "overdue:remind";

    private static final String OVERDUE_TASK_REVERSE_KEY = "overdue:tasks:reverse";

    private static final String OVERDUE_REMIND_REVERSE_KEY = "overdue:remind:reverse";

    private final ApplicationEventPublisher eventPublisher;
    private static final String LOCK_PREFIX = "timeout:lock:";
    private static final int LOCK_EXPIRE_SECONDS = 120; // 2分钟
    public RedisDelayQueue(ApplicationEventPublisher eventPublisher) {
        this.eventPublisher = eventPublisher;
    }
    /**
     * 添加超时任务-支持实例ID
     */
    public void addTimeoutTask(Long instanceId,String bizType, Long bizId, LocalDateTime executeTime) {
        try {
            long timestamp = executeTime.toInstant(ZoneOffset.of("+8")).toEpochMilli();

            Map<String, Object> taskData = new HashMap<>();
            taskData.put("instanceId", instanceId);
            taskData.put("bizType", bizType);
            taskData.put("bizId", bizId);
            taskData.put("type", "timeout");
            taskData.put("executeTime", executeTime.toString());

            String taskJson = JSON.toJSONString(taskData);

            // 添加到有序集合
            redisTemplate.opsForZSet().add(TIMEOUT_QUEUE_KEY, taskJson, timestamp);

            // 添加映射关系，便于删除
            redisTemplate.opsForValue().set(TASK_MAPPING_KEY + "timeout:" + instanceId, taskJson);

            log.info("添加超时任务成功: instanceId={}, executeTime={}", instanceId, executeTime);
        } catch (Exception e) {
            log.error("添加超时任务失败: instanceId={}", instanceId, e);
        }
    }

    /**
     * 添加提醒任务-支持实例ID
     */
    public void addRemindTask(Long instanceId,String bizType, Long bizId,  LocalDateTime executeTime) {
        try {
            long timestamp = executeTime.toInstant(ZoneOffset.of("+8")).toEpochMilli();

            Map<String, Object> taskData = new HashMap<>();
            taskData.put("instanceId", instanceId);
            taskData.put("bizType", bizType);
            taskData.put("bizId", bizId);
            taskData.put("type", "remind");
            taskData.put("executeTime", executeTime.toString());

            String taskJson = JSON.toJSONString(taskData);

            // 添加到有序集合
            redisTemplate.opsForZSet().add(REMIND_QUEUE_KEY, taskJson, timestamp);

            // 添加映射关系，便于删除
            redisTemplate.opsForValue().set(TASK_MAPPING_KEY + "remind:" + instanceId, taskJson);

            log.info("添加提醒任务成功: instanceId={}, executeTime={}", instanceId, executeTime);
        } catch (Exception e) {
            log.error("添加提醒任务失败: instanceId={}", instanceId, e);
        }
    }

    /**
     * 移除超时任务
     */
    public void removeTimeoutTask(Long instanceId) {
        try {
            String mappingKey = TASK_MAPPING_KEY + "timeout:" + instanceId;
            String taskJson = (String) redisTemplate.opsForValue().get(mappingKey);

            if (taskJson != null) {
                // 从有序集合中移除
                redisTemplate.opsForZSet().remove(TIMEOUT_QUEUE_KEY, taskJson);
                // 删除映射关系
                redisTemplate.delete(mappingKey);
                log.info("移除超时任务成功: instanceId={}", instanceId);
            } else {
                log.info("超时任务不存在: instanceId={}", instanceId);
            }
        } catch (Exception e) {
            log.error("移除超时任务失败: instanceId={}", instanceId, e);
        }
    }

    /**
     * 移除提醒任务
     */
    public void removeRemindTask(Long instanceId) {
        try {
            String mappingKey = TASK_MAPPING_KEY + "remind:" + instanceId;
            String taskJson = (String) redisTemplate.opsForValue().get(mappingKey);

            if (taskJson != null) {
                // 从有序集合中移除
                redisTemplate.opsForZSet().remove(REMIND_QUEUE_KEY, taskJson);
                // 删除映射关系
                redisTemplate.delete(mappingKey);
                log.info("移除提醒任务成功: instanceId={}", instanceId);
            } else {
                log.info("提醒任务不存在: instanceId={}", instanceId);
            }
        } catch (Exception e) {
            log.error("移除提醒任务失败: instanceId={}", instanceId, e);
        }
    }

    /**
     * 移除实例的所有任务
     */
    public void removeAllTasksForInstance(Long instanceId) {
        removeTimeoutTask(instanceId);
        removeRemindTask(instanceId);
    }
    /**
     * 获取即将到期的超时任务
     */
    public Set<Object> getExpiredTimeoutTasks() {
        try {
            long currentTimestamp = System.currentTimeMillis();
            return redisTemplate.opsForZSet().rangeByScore(TIMEOUT_QUEUE_KEY, 0, currentTimestamp);
        } catch (Exception e) {
            log.error("获取即将到期的超时任务失败", e);
            return null;
        }
    }

    /**
     * 获取即将到期的提醒任务
     */
    public Set<Object> getExpiredRemindTasks() {
        try {
            long currentTimestamp = System.currentTimeMillis();
            log.info("获取系统当前时间={}",currentTimestamp);
            return redisTemplate.opsForZSet().rangeByScore(REMIND_QUEUE_KEY, 0, currentTimestamp);
        } catch (Exception e) {
            log.error("获取即将到期的提醒任务失败", e);
            return null;
        }
    }

    /**
     * 处理已到期的超时任务
     */
    public void processExpiredTimeoutTasks() {
        String lockKey = LOCK_PREFIX + "process_timeout";
        String lockValue = UUID.randomUUID().toString();

        Boolean locked = redisTemplate.opsForValue()
            .setIfAbsent(lockKey, lockValue, Duration.ofSeconds(LOCK_EXPIRE_SECONDS));

        if (Boolean.FALSE.equals(locked)) {
            log.debug("其他实例正在处理超时任务，跳过本次执行");
            return;
        }

        try {
            log.info("=== 开始批量处理超时任务 ===");

            long currentTimestamp = System.currentTimeMillis();
            int batchSize = 100; // 每批处理100个任务
            int totalProcessed = 0;

            while (true) {
                // 获取一批任务
                Set<Object> batchTasks = redisTemplate.opsForZSet()
                    .rangeByScore(TIMEOUT_QUEUE_KEY, 0, currentTimestamp, 0, batchSize);

                if (batchTasks == null || batchTasks.isEmpty()) {
                    break;
                }

                log.info("处理第 {} 批任务，数量: {}", (totalProcessed / batchSize) + 1, batchTasks.size());

                // 原子性移除这批任务
                for (Object taskObj : batchTasks) {
                    redisTemplate.opsForZSet().remove(TIMEOUT_QUEUE_KEY, taskObj);
                }

                // 处理任务
                for (Object taskObj : batchTasks) {
                    try {
                        //处理超时事件
                        processTimeoutTask(taskObj);
                        totalProcessed++;
                    } catch (Exception e) {
                        log.error("处理超时任务失败: {}", taskObj, e);
                    }
                }

                // 如果处理的任务数少于批次大小，说明已经处理完了
                if (batchTasks.size() < batchSize) {
                    break;
                }
            }
            log.info("=== 超时任务处理完成，总计处理: {} 个任务 ===", totalProcessed);
        } finally {
            releaseLock(lockKey, lockValue);
        }
    }
    /**
     * 处理单个超时任务
     */
    private void processTimeoutTask(Object taskObj) {
        try {
            @SuppressWarnings("unchecked")
            Map<String, Object> taskData = JSON.parseObject(taskObj.toString(), Map.class);
            Long instanceId = Long.valueOf(taskData.get("instanceId").toString());
            log.info("进入到【处理单个超时任务】instanceId ={} 开始",instanceId);
            // 删除映射关系
            redisTemplate.delete(TASK_MAPPING_KEY + "timeout:" + instanceId);

            // 发布事件前先检查实例是否还存在且为WAITING状态
            if (isInstanceValidForProcessing(instanceId)) {
                publishTimeoutEvent(instanceId);
                log.info("✓ 进入到【处理单个超时任务】: instanceId={} 结束", instanceId);
            } else {
                log.warn("实例状态已变更，跳过处理: instanceId={}", instanceId);
            }

        } catch (Exception e) {
            log.error("处理超时任务异常: {}", taskObj, e);
            throw e;
        }
    }

    /**
     * 检查实例是否有效且可处理
     */
    private boolean isInstanceValidForProcessing(Long instanceId) {
        try {
            // 这里需要调用你的service检查实例状态
            TimeoutInstance instance = timeoutInstanceService.getById(instanceId);
            return instance != null && "WAITING".equals(instance.getStatus());
        } catch (Exception e) {
            log.error("检查实例状态失败: instanceId={}", instanceId, e);
            return false;
        }
    }
    /**
     * 处理已到期的提醒任务
     */
    public void processExpiredRemindTasks() {
        try {
            log.info("进入到【到期的提醒任务】processExpiredRemindTasks");
            Set<Object> expiredTasks = getExpiredRemindTasks();
            if (expiredTasks != null && !expiredTasks.isEmpty()) {
                for (Object taskObj : expiredTasks) {
                    try {
                        @SuppressWarnings("unchecked")
                        Map<String, Object> taskData = JSON.parseObject(taskObj.toString(), Map.class);
                        if (StringUtils.isNotBlank(String.valueOf(taskData.get("instanceId")))) {
                            Long instanceId = Long.valueOf(taskData.get("instanceId").toString());
                            log.info("进入到【到期的提醒任务】instanceId =={} 开始",instanceId);
                            // 从队列中移除已处理的任务
                            redisTemplate.opsForZSet().remove(REMIND_QUEUE_KEY, taskObj);

                            // 删除映射关系
                            redisTemplate.delete(TASK_MAPPING_KEY + "remind:" + instanceId);

                            // 发布提醒事件
                            publishRemindEvent(instanceId);

                            log.info("进入到【到期的提醒任务】: instanceId={} 结束", instanceId);
                        }
                    } catch (Exception e) {
                        log.error("处理单个提醒任务失败: {}", taskObj, e);
                    }
                }
            }
        } catch (Exception e) {
            log.error("处理已到期的提醒任务失败", e);
        }
    }

    /**
     * 发布超时事件（可以通过Spring Events或消息队列实现）
     */
    private void publishTimeoutEvent(Long instanceId) {
        log.info("发布超时事件: instanceId={}", instanceId);
        eventPublisher.publishEvent(new TimeoutEvent(this,instanceId));
    }

    /**
     * 发布提醒事件
     */
    private void publishRemindEvent(Long instanceId) {
        log.info("发布提醒事件: instanceId={}", instanceId);
        eventPublisher.publishEvent(new RemindEvent(this,instanceId));
    }

    /**
     * 获取队列中的任务数量
     */
    public Map<String, Long> getQueueSize() {
        Map<String, Long> queueSize = new HashMap<>();
        try {
            queueSize.put("timeoutTasks", redisTemplate.opsForZSet().zCard(TIMEOUT_QUEUE_KEY));
            queueSize.put("remindTasks", redisTemplate.opsForZSet().zCard(REMIND_QUEUE_KEY));
        } catch (Exception e) {
            log.error("获取队列大小失败", e);
            queueSize.put("timeoutTasks", 0L);
            queueSize.put("remindTasks", 0L);
        }
        return queueSize;
    }

    /**
     * 清理所有队列（谨慎使用）
     */
    public void clearAllQueues() {
        try {
            redisTemplate.delete(TIMEOUT_QUEUE_KEY);
            redisTemplate.delete(REMIND_QUEUE_KEY);

            // 清理映射关系
            Set<String> mappingKeys = redisTemplate.keys(TASK_MAPPING_KEY + "*");
            if (mappingKeys != null && !mappingKeys.isEmpty()) {
                redisTemplate.delete(mappingKeys);
            }

            log.info("清理所有队列成功");
        } catch (Exception e) {
            log.error("清理所有队列失败", e);
        }
    }

    // ========== ========================================兼容 响应 超期的方法 ==========

    /**
     * 添加超期任务
     */
    public void addTask(String bizType, Long bizId, LocalDateTime executeTime) {
        // 1. 唯一键
        String memberKey = UUID.randomUUID().toString();

        // 2. 确定性 JSON（顺序 + 无空格 + 无转义）
        Map<String, Object> map = new LinkedHashMap<>();
        map.put("bizType", bizType);
        map.put("bizId", bizId);
        map.put("type", "overdue");
        map.put("memberKey", memberKey);

        // **关键：先转 JSON，再 parse 一次，去掉转义斜杠**
        String taskValue = JSON.parseObject(JSON.toJSONString(map)).toString();

        // 3. 时间戳
        long timestamp = executeTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();

        // 4. 写入
        redisTemplate.opsForZSet().add(OVERDUE_TASK_KEY, taskValue, timestamp);
        redisTemplate.opsForHash().put(OVERDUE_TASK_REVERSE_KEY, memberKey, taskValue);

        log.info(">>> addTask ok: memberKey={}, bizType={}, bizId={}, timestamp={}",
            memberKey, bizType, bizId, timestamp);
    }

    /**
     * 添加提醒任务
     */
    public void addRemindTask(String bizType, Long bizId, LocalDateTime executeTime) {
        try {
            // 1. 唯一键
            String memberKey = UUID.randomUUID().toString();

            // 2. 确定性 JSON（顺序 + 无空格 + 无转义）
            Map<String, Object> map = new LinkedHashMap<>();
            map.put("bizType", bizType);
            map.put("bizId", bizId);
            map.put("type", "overdue");
            map.put("memberKey", memberKey);

            // **关键：先转 JSON，再 parse 一次，去掉转义斜杠**
            String taskValue = JSON.parseObject(JSON.toJSONString(map)).toString();

            // 3. 时间戳
            long timestamp = executeTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();

            // 4. 写入
            redisTemplate.opsForZSet().add(OVERDUE_REMIND_KEY, taskValue, timestamp);
            redisTemplate.opsForHash().put(OVERDUE_REMIND_REVERSE_KEY, memberKey, taskValue);

            log.info(">>> addTask ok: memberKey={}, bizType={}, bizId={}, timestamp={}",
                memberKey, bizType, bizId, timestamp);

        } catch (Exception e) {
            log.error("添加提醒任务失败: bizType={}, bizId={}", bizType, bizId, e);
        }
    }

    /**
     * 移除响应  超期任务
     */
    public void removeTask(String bizType, Long bizId) {
        try {
            Set<String> memberKeys = getMemberKeys(OVERDUE_TASK_REVERSE_KEY,bizType, bizId);
            for (String mk : memberKeys) {
                String taskValue = (String) redisTemplate.opsForHash().get(OVERDUE_TASK_REVERSE_KEY, mk);
                if (taskValue != null) {
                    redisTemplate.opsForZSet().remove(OVERDUE_TASK_KEY, taskValue);
                    redisTemplate.opsForHash().delete(OVERDUE_TASK_REVERSE_KEY, mk);
                }
            }
            log.info(">>> removeTask ok: bizType={}, bizId={}, delCount={}", bizType, bizId, memberKeys.size());
        } catch (Exception e) {
            log.error("移除超期任务失败: bizType={}, bizId={}", bizType, bizId, e);
        }
    }

    /**
     * 移除响应 提醒任务
     */
    public void removeRemindTask(String bizType, Long bizId) {
        try {
            Set<String> memberKeys = getMemberKeys(OVERDUE_REMIND_REVERSE_KEY,bizType, bizId);

            for (String mk : memberKeys) {
                String taskValue = (String) redisTemplate.opsForHash().get(OVERDUE_REMIND_REVERSE_KEY, mk);
                if (taskValue != null) {
                    redisTemplate.opsForZSet().remove(OVERDUE_REMIND_KEY, taskValue);
                    redisTemplate.opsForHash().delete(OVERDUE_REMIND_REVERSE_KEY, mk);
                }
            }
            log.info(">>> removeRemindTask ok: bizType={}, bizId={}, delCount={}", bizType, bizId, memberKeys.size());
        } catch (Exception e) {
            log.error("移除提醒任务失败: bizType={}, bizId={}", bizType, bizId, e);
        }
    }

    @NotNull
    private Set<String> getMemberKeys(String key,String bizType, Long bizId) {
        Set<String> memberKeys = redisTemplate.opsForHash().keys(key)
            .stream()
            .map(Object::toString)
            .filter(k -> {
                String json = (String) redisTemplate.opsForHash().get(key, k);
                Map<?,?> m = JSON.parseObject(json, Map.class);
                // 1. 先判断 bizType 匹配
                if (!bizType.equals(m.get("bizType"))) {
                    return false;
                }
                // 2. 统一 bizId 为 Long 类型比较
                Object bizIdObj = m.get("bizId");
                if (bizIdObj == null) {
                    return false;
                }
                Long parsedBizId = null;
                if (bizIdObj instanceof Integer) {
                    parsedBizId = ((Integer) bizIdObj).longValue(); // Integer 转 Long
                } else if (bizIdObj instanceof Long) {
                    parsedBizId = (Long) bizIdObj;
                } else if (bizIdObj instanceof String) {
                    parsedBizId = Long.parseLong((String) bizIdObj); // 字符串转 Long
                }
                return bizId.equals(parsedBizId); // 用 Long 比较
            })
            .collect(Collectors.toSet());
        return memberKeys;
    }


    /**
     * 释放分布式锁的辅助方法
     */
    private void releaseLock(String lockKey, String lockValue) {
        try {
            String unlockScript =
                "if redis.call('GET', KEYS[1]) == ARGV[1] then\n" +
                    "    return redis.call('DEL', KEYS[1])\n" +
                    "else\n" +
                    "    return 0\n" +
                    "end";

            Long result = redisTemplate.execute(
                new DefaultRedisScript<>(unlockScript, Long.class),
                Collections.singletonList(lockKey),
                lockValue
            );

            if (result != null && result == 1L) {
                log.debug("成功释放锁: {}", lockKey);
            } else {
                log.warn("锁可能已过期或被其他实例释放: {}", lockKey);
            }
        } catch (Exception e) {
            log.error("释放锁失败: {}", lockKey, e);
        }
    }
}
